package com.bins.redis.distributedlock;

import java.nio.charset.Charset;
import java.util.UUID;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.RedisStringCommands;
import org.springframework.data.redis.connection.ReturnType;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.types.Expiration;
import org.springframework.stereotype.Component;

@Component
public class RedisDistributedLock extends AbstractDistributedLock {

	@Autowired
	private RedisTemplate<Object, Object> redisTemplate;

	private final Logger logger = LoggerFactory.getLogger(RedisDistributedLock.class);

	private ThreadLocal<String> lockFlag = new ThreadLocal<String>();

	public static final String UNLOCK_LUA;

	static {
		StringBuilder sb = new StringBuilder();
		sb.append("if redis.call(\"get\",KEYS[1]) == ARGV[1] ");
		sb.append("then ");
		sb.append("    return redis.call(\"del\",KEYS[1]) ");
		sb.append("else ");
		sb.append("    return 0 ");
		sb.append("end ");
		UNLOCK_LUA = sb.toString();
	}

	public RedisDistributedLock() {
		super();
	}

	@Override
	public boolean lock(String key, long expire, int retryTimes, long sleepMillis) {
		boolean result = setRedis(key, expire);
		while ((!result) && retryTimes-- > 0) {
			try {
				logger.debug("lock failed, retrying..." + retryTimes);
				Thread.sleep(sleepMillis);
			} catch (InterruptedException e) {
				return false;
			}
			result = setRedis(key, expire);
		}
		return result;
	}

	private boolean setRedis(final String key, final long expire) {
		try {
			RedisCallback<Boolean> callback = (connection) -> {
				String uuid = UUID.randomUUID().toString();
				lockFlag.set(uuid);
				return connection.set(key.getBytes(Charset.forName("UTF-8")), uuid.getBytes(Charset.forName("UTF-8")),
						Expiration.milliseconds(expire), RedisStringCommands.SetOption.SET_IF_ABSENT);
			};
			return (Boolean) redisTemplate.execute(callback);
		} catch (Exception e) {
			logger.error("redis lock error.", e);
		}
		return false;
	}

	@Override
	public boolean releaseLock(String key) {
		try {
			RedisCallback<Boolean> callback = (connection) -> {
				String value = lockFlag.get();
				return connection.eval(UNLOCK_LUA.getBytes(), ReturnType.BOOLEAN, 1,
						key.getBytes(Charset.forName("UTF-8")), value.getBytes(Charset.forName("UTF-8")));
			};
			return (Boolean) redisTemplate.execute(callback);
		} catch (Exception e) {
			logger.error("release lock occured an exception", e);
		} finally {
			lockFlag.remove();
		}
		return false;
	}

}